package com.asiainfo.dacp.datastash.broker.facade.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.core.BrokerModel;
import com.asiainfo.dacp.datastash.broker.core.StashUnitProcessModel;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitTaskStateEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.Broker;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.facade.state.UnitTaskStateLogService;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogVo;
import com.asiainfo.dacp.datastash.broker.tracelog.StashTraceLogger;

/**
 * 任务派发类
 *
 * 任务派发需要完成对于broker状态的判断，因为stash主要做elt属于网络、IO密集的操作， 一、任务分派原则
 * 1.不分割任务：分配的原则是找到一台IO，网络最闲的机器完成操作，使整个集群的IO，网络尽量跑满 2.分割任务：暂时不实现，需要对于unit的分割抽象
 *
 * 二、任务优先级 1.同时到达多个任务是，需要完成任务优先级的判断，来分配那个任务先启动
 *
 * 二、任务容错原则 1.断点续传： 2.任务重做： 3.
 *
 * @author MeiKefu
 * @date 2017-10-31
 */
@Service
public class UnitDispatcher {

	@Autowired
	private BrokerModel brokerFacade;

	@Autowired
	private StashUnitProcessModel stashUnitProcessModel;

	@Autowired
	private BrokerModel brokerModel;

	@Autowired
	private LinearProgramming linearProgramming;

	@Autowired
	private UnitTaskStateLogService taskLog;

	@Autowired
	private StashTraceLogger tracelog;
	private String clzName = this.getClass().getSimpleName();

	public void dispatch(StashUnitTask stashUnitTask) {

		tracelog.info(StashTraceLogVo.buildStashTraceLogVo(stashUnitTask, clzName, "dispatch", "开始分配unitTask"));

		// 规划好的分片与主机
		Broker broker = linearProgramming.programming(stashUnitTask, brokerFacade.brokers());

		if (broker != null) {
			stashUnitTask.setBrokerId(broker.getBrokerId());// TODO
															// stashUnitTask里面broker没有更新
			stashUnitTask.runnable();

			taskLog.update(stashUnitTask, UnitTaskStateEnum.RUNNABLE, broker.getBrokerId(), broker.getBrokerIp());

			brokerModel.pushWork(stashUnitTask);
			brokerModel.putBrokerUnitTaskList(stashUnitTask.getTaskProcId(), broker.getBrokerId());
		} else {
			stashUnitProcessModel.requeueUnitTaskDispatchQueue(stashUnitTask);
		}
	}

}
